Description

This PR introduces a lazy approach to handling potentially corrupted indices. Previously only recoverIndex API has such capability introduced in #241 and this PR makes this logic generic across all Flint APIs.

A corrupted index is defined as a Flint index has an index log entry but no corresponding data index. Upon the next Flint API call, the system will check for such corrupted indices and perform the necessary cleanup, ensuring that users can recreate indices without encountering errors.

Typical Scenario

The most common scenario this PR aims to address is:

1. Users delete OS data index for a Flint index directly
2. Index monitor terminates streaming job if auto refresh enabled
3. Users attempt to create index again
   a. Before changes: the creation failed and require to remove the log entry manually
   b. After changes: the log entry is cleaned up automatically and creation succeeds

Uncommon Scenarios and Handling

After step 1 above, user may rarely attempt to:

1. Create or vacuum index again concurrently: Skip corruption check if index in CREATING or VACUUMIGN state to reduce the possibility of race condition, ensuring log entry won't be removed mistakenly during ongoing operation.
2. Operate before index monitor terminates streaming job: If index corrupted, skip heartbeat reporting in index monitor which relies on log entry to reduce potential conflict.

Note: The changes aim to reduce the possibility of race conditions with best efforts. However, due to the lack of transaction support in OpenSearch, any checks performed before acquiring an optimistic lock are still subject to "dirty reads." This means there remains a small possibility of inconsistencies during concurrent operations.

Example

Prepare corrupted Flint index:

# Create an auto refresh Flint index
CREATE SKIPPING INDEX ON glue.default.http_logs (
  year PARTITION
)
WITH (
  auto_refresh = true,
  checkpoint_location = 's3://checkpoint-1'
);

# Delete OS index once all data refreshed
DELETE flint_glue_default_http_logs_skipping_index

# Index monitor detects and terminates streaming job
24/09/30 18:27:43 WARN FlintSparkIndexMonitor: Streaming job is active but data is deleted
24/09/30 18:27:43 INFO FlintSparkIndexMonitor: Terminating streaming job and index monitor for
 flint_glue_default_http_logs_skipping_index

# Flint index log entry is dangling in failed state
GET .query_execution_request_glue/_search
          "version": "1.0",
          "latestId": "ZmxpbnRfZ2x1ZV9kZWZhdWx0X2h0dHBfbG9nc19za2lwcGluZ19pbmRleA==",
          "type": "flintindexstate",
          "state": "failed",
          "applicationId": "XXX",
          "jobId": "YYY",
          "dataSourceName": "glue",
          "jobStartTime": 1727720050924,
          "lastUpdateTime": 1727720863284,
          "error": ""
        }

Verify cleanup logic works:

# Attempt to create the same index again
CREATE SKIPPING INDEX ON glue.default.http_logs (
  year PARTITION
)
WITH (
  auto_refresh = true,
  checkpoint_location = 's3://checkpoint-2'
);

# Index metadata log entry is cleaned up
24/09/30 18:34:35 INFO FlintSpark: Starting index operation 
[Create Flint index flint_glue_default_http_logs_skipping_index] with forceInit=true
24/09/30 18:34:35 WARN FlintSpark: 
 Cleaning up corrupted index:
 - logEntryExists [true]
 - dataIndexExists [false]
 - isCreatingOrVacuuming [false]
24/09/30 18:34:39 INFO FlintSpark: Index operation
[Create Flint index flint_glue_default_http_logs_skipping_index] complete

GET .query_execution_request_glue/_search
          "version": "1.0",
          "latestId": "ZmxpbnRfZ2x1ZV9kZWZhdWx0X2h0dHBfbG9nc19za2lwcGluZ19pbmRleA==",
          "type": "flintindexstate",
          "state": "refreshing",
          "applicationId": "XXX",
          "jobId": "YYY",
          "dataSourceName": "glue",
          "jobStartTime": 1727721279398,
          "lastUpdateTime": 1727723128616,
          "error": ""
        }

Issues Resolved

#356

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.
For more information on following Developer Certificate of Origin and signing off your commits, please check here.